package com.zyzg.listenter;

import com.alibaba.fastjson2.JSONObject;
import com.rabbitmq.client.Channel;
import com.zyzg.constant.BusinessConstants;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * 数据监听类
 * @author rui
 */
@Slf4j
@Component
public class BusinessMsgListener {

    private static final ConcurrentLinkedQueue<JSONObject> queue_0200 = new ConcurrentLinkedQueue<>();
    private static final ConcurrentLinkedQueue<JSONObject> queue_0704 = new ConcurrentLinkedQueue<>();


    // 提供公共方法获取 queue_0200 队列
    public ConcurrentLinkedQueue<JSONObject> getQueue_0200() {
        return queue_0200;
    }

    // 提供公共方法获取 queue_0704 队列
    public ConcurrentLinkedQueue<JSONObject> getQueue_0704() {
        return queue_0704;
    }

    /**
     * 设备位置数据消费
     * @param message
     * @return
     */
    @RabbitHandler
    @RabbitListener(bindings = {@QueueBinding(
            value = @Queue(value = BusinessConstants.JT808_LOCATION_QUEUE, durable = "true", declare = "true"),
            exchange = @Exchange(value = BusinessConstants.JT808_LOCATION_EXCHANGE, type = "fanout", durable = "true", declare = "true"),
            declare = "true")}, containerFactory = "businessContainerFactory")
    public void processLocation(Message message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        try {
            String msgStr = new String(message.getBody(), StandardCharsets.UTF_8);
            // log.info("BusinessMsgListener processLocation() 监听到一条消息:{}", msgStr);
            JSONObject msg = JSONObject.parseObject(msgStr);
            // 剩下业务逻辑处理
            // 判断type字段，进行分类处理
            String msgType = msg.getString("type");

            processMessageAsync(msg, msgType, channel, deliveryTag);

        } catch (Exception e) {
            try {
                // 消息处理失败，拒绝消息并重新入队
                channel.basicNack(deliveryTag, false, true);
            } catch (Exception nackException) {
                nackException.printStackTrace();
            }
            e.printStackTrace();
        }
    }

    @Async("messageProcessingExecutor")
    public void processMessageAsync(JSONObject msg, String msgType, Channel channel, long deliveryTag) {
        try {
            // 向缓存队列中添加
            if (msgType.equals("0x0200")) {
                queue_0200.offer(msg);
            } else if (msgType.equals("0x0704")) {
                queue_0704.offer(msg);
            }
            // 消息处理成功，手动确认消息
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            try {
                channel.basicNack(deliveryTag, false, true);
            } catch (Exception nackException) {
                nackException.printStackTrace();
            }
            e.printStackTrace();
        }
    }

}
